package com.sh.rpc.registry;

import com.sh.rpc.confige.RegistryConfig;
import com.sh.rpc.model.ServiceMetaInfo;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 基于zookeeper的注册中心
 *
 * @author sh
 */
public class ZookeeperRegistry implements Registry {

    /**
     * Curator 是一个用于简化与 ZooKeeper 交互的 Java 客户端库  比原生简单，也可以和zookeeper通信
     */
    private CuratorFramework client;   // zookeeper客户端
    private ServiceDiscovery<ServiceMetaInfo> serviceDiscovery;  // 服务发现

    public static final Set<String> localRegisterNodeSet = new HashSet<>();  // 本地注册节点集合
    private RegistryServiceCache registryServiceCache = new RegistryServiceCache();  // 注册中心服务缓存
    private final Set<String> watchingKeySet = new HashSet<>();  // 正在监听的key集合
    /**
     * 根节点
     * 这是向 zookeeper 注册的根节点
     */
    private static final String ZOOKEEPER_ROOT_PATH = "/rpc/zk";


    @Override
    public void init(RegistryConfig registryConfig) {
//        // 创建 CuratorFramework 客户端
        client = CuratorFrameworkFactory
                .builder()
                // 设置连接地址，从 registryConfig 对象中获取
                .connectString(registryConfig.getAddress())
                // 设置重试策略
                .retryPolicy(new ExponentialBackoffRetry(Math.toIntExact(registryConfig.getTimeout()), 3))
                // 完成构建并返回 CuratorFramework 对象
                .build();

        // 初始化 serviceDiscovery 对象，用于服务发现
        serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMetaInfo.class)
                // 设置 CuratorFramework 客户端，用于与 Zookeeper 通信
                .client(client)
                // 设置 Zookeeper 中的基础路径
                .basePath(ZOOKEEPER_ROOT_PATH)
                // 设置序列化器，用于将 ServiceMetaInfo 对象序列化为 JSON 格式存储在 Zookeeper 中
                .serializer(new JsonInstanceSerializer<>(ServiceMetaInfo.class))
                // 完成构建并返回 ServiceDiscovery 对象
                .build();

        try {
            client.start();  // 启动zookeeper客户端
            serviceDiscovery.start();  // 启动服务发现
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 向服务中心 etcd 注册服务信息
     *
     * @param serviceMeteInfo
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMeteInfo) throws Exception {
        // 注册到zookeeper
        serviceDiscovery.registerService(buildServiceInstance(serviceMeteInfo));

        // 将服务元信息添加到本地注册节点集合中
        String registerKey = ZOOKEEPER_ROOT_PATH + "/" + serviceMeteInfo.getServiceNodeKey();
        localRegisterNodeSet.add(registerKey);
    }

    /**
     * 取消注册 服务端
     */
    @Override
    public void unRegister(ServiceMetaInfo serviceMeteInfo) throws Exception {
        try {
            // 取消注册服务
            serviceDiscovery.unregisterService(buildServiceInstance(serviceMeteInfo));
            System.out.println(serviceMeteInfo.getServiceName()+"=====>取消成功");
        } catch (Exception e) {
            throw new RuntimeException( serviceMeteInfo.getServiceName()+"=====>取消错误" , e);
        }

        // 从本地注册节点集合中移除对应的注册节点
        String registerKey = ZOOKEEPER_ROOT_PATH + "/" + serviceMeteInfo.getServiceNodeKey();
        localRegisterNodeSet.remove(registerKey);

    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        return null;
    }

    @Override
    public void destroy() {

    }

    @Override
    public void heartbeat() {

    }

    @Override
    public void watch(String serviceNodeKey) {

    }


    private ServiceInstance<ServiceMetaInfo> buildServiceInstance(ServiceMetaInfo serviceMetaInfo) {
        // 根据服务元信息中的服务主机地址和端口号拼接出服务地址
        String serviceAddress = serviceMetaInfo.getServiceHost() + ":" + serviceMetaInfo.getServicePort();
        try {
            // 创建一个 ServiceInstance 对象的构建器，并进行链式调用设置属性
            return ServiceInstance.<ServiceMetaInfo>builder()
                    // 设置服务实例的唯一标识为拼接后的服务地址
                    .id(serviceAddress)
                    // 设置服务名称为服务元信息中的服务键
                    .name(serviceMetaInfo.getServiceKey())
                    // 设置服务地址为拼接后的服务地址
                    .address(serviceAddress)
                    // 设置服务实例的负载信息为传入的服务元信息对象
                    .payload(serviceMetaInfo)
                    // 完成构建并返回构建好的 ServiceInstance 对象
                    .build();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
